package cn.datawin.cmppclient.sender.handler;

import cn.datawin.cmppclient.sender.service.ProducerService;
import cn.datawin.cmppclient.sender.thread.CommonThreadPoolFactory;
import cn.datawin.constant.RocketMqConstant;
import cn.datawin.sms.cmpp2.ReportCache;
import cn.datawin.sms.cmpp2.ReportLog;
import cn.datawin.sms.cmpp2.SmsDetailLog;
import cn.datawin.sms.cmpp2.SubmitRes;
import cn.hutool.extra.spring.SpringUtil;
import com.alibaba.fastjson.JSON;
import com.drondea.sms.message.IMessage;
import com.drondea.sms.message.cmpp.CmppSubmitRequestMessage;
import com.drondea.sms.message.cmpp.CmppSubmitResponseMessage;
import com.drondea.sms.type.IMessageResponseHandler;
import lombok.extern.slf4j.Slf4j;
import org.joda.time.DateTime;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;

import java.util.Arrays;
import java.util.Date;
import java.util.List;

import static cn.datawin.sms.enums.ApiStatus._40001;


@Slf4j
public class CmppMessageResponseHandler implements IMessageResponseHandler {

    private SmsDetailLog smsDetailLog;

    public CmppMessageResponseHandler(SmsDetailLog smsDetailLog) {
        this.smsDetailLog = smsDetailLog;
    }

    @Override
    public void messageComplete(IMessage msgRequest, IMessage msgResponse) {
//        log.debug("messageComplete--------------------");
        msgResponse.setRequestMessage(msgRequest);
        //1、设置回更的msgid，丢入更新队列
        CmppSubmitResponseMessage cmppSubmitResponseMessage = (CmppSubmitResponseMessage) msgResponse;
        CmppSubmitRequestMessage cmppSubmitRequestMessage = (CmppSubmitRequestMessage) msgRequest;
        String prMsgId = cmppSubmitResponseMessage.getMsgId().toString();
        long result = cmppSubmitResponseMessage.getResult();
//        log.debug("messageComplete,submit={},prMsgId={}", JSON.toJSONString(submit), prMsgId);
        List<String> crMsgIds = Arrays.asList(smsDetailLog.getCrMsgId().split(","));


        SubmitRes submitRes = new SubmitRes();
        submitRes.setSmsDetailLogId(smsDetailLog.getId());
        submitRes.setCreateDate(new Date());
        submitRes.setCrMsgId(smsDetailLog.getCrMsgId());
        submitRes.setPrMsgId(prMsgId);
        submitRes.setSubmitResult((int) result);
        submitRes.setSubmitStatus(0);
        submitRes.setYmd(smsDetailLog.getYmd());
        submitRes.setSeq(cmppSubmitResponseMessage.getSequenceId());
        submitRes.setPkTotal(cmppSubmitRequestMessage.getPkTotal());
        submitRes.setPkNumber(cmppSubmitRequestMessage.getPkNumber());
        ProducerService producerService = SpringUtil.getBean(ProducerService.class);
        log.debug("submitRes,{}", JSON.toJSONString(submitRes));
        producerService.sendDbMessage(submitRes);
//        producerService.sendMessage(RocketMqCommonTopic.PROCESS_LOG_TOPIC + ":" + SubmitRes.COLLECTION_NAME, JSON.toJSONString(submitRes));

		//TODO
        //发送响应更新到统计表中
        /*SendDetailStat sendDetail = new SendDetailStat();
        sendDetail.setSendDetailId(submitLog.getId());
        sendDetail.setSubmitStatus(1);
        sendDetail.setYmd(submitLog.getYmd());
//        producerService.sendOrderMessage(RocketMqCommonTopic.SEND_DETAIL_TOPIC, JSON.toJSONString(sendDetail),submitLog.getId());
//        producerService.sendMessage(RocketMqCommonTopic.SEND_DETAIL_TOPIC+"_tmp",JSON.toJSONString(sendDetail));
        producerService.sendDbMessage(sendDetail,SendDetailStat.COLLECTION_NAME+"_tmp");*/


        //2、设置prMsgid -> crMsgid的缓存，72小时

        ReportCache reportCache = new ReportCache();
        reportCache.setSmsDetailLogId(smsDetailLog.getId());
        reportCache.setPrMsgId(prMsgId);
		reportCache.setCstmCustomerId(smsDetailLog.getCstmCustomerId());
		reportCache.setCstmCustomerCode(smsDetailLog.getCstmCustomerCode());
		reportCache.setCreateDate(new Date());
        reportCache.setYmd(smsDetailLog.getYmd());
//        reportCache.setOriginCrMsgId(smsDetailLog.getOriginCrMsgId());
//        reportCache.setOriginId(smsDetailLog.getOriginId());
        reportCache.setChannelId(Long.valueOf(smsDetailLog.getProvProviderChannelId()));
        short pkNumber = submitRes.getPkNumber();
        String singleCrMsgId = crMsgIds.get(pkNumber-1);
        reportCache.setCrMsgIds(singleCrMsgId);
        reportCache.setAllCrMsgIds(smsDetailLog.getCrMsgId());
        reportCache.setMsgSize(crMsgIds.size());
        producerService.sendDbMessage(reportCache);

        try {
            if(result!=0) {
				MongoTemplate mongoHelper = SpringUtil.getBean(MongoTemplate.class);
                ReportLog reportLog = new ReportLog();
                reportLog.setSmsDetailLogId(smsDetailLog.getId());
                reportLog.setCrMsgId(smsDetailLog.getCrMsgId());
                reportLog.setStat("REJECT:" + result);
                reportLog.setSubmitTime(new DateTime(smsDetailLog.getSubmitDate()).toString("yyMMddHHmm"));
                reportLog.setDoneTime(new DateTime().toString("yyMMddHHmm"));
                reportLog.setPhoneNo(smsDetailLog.getPhoneNos());
                reportLog.setCreateDate(new Date());
                reportLog.setPrMsgId(prMsgId);
                reportLog.setProviderNo(smsDetailLog.getProvProviderId());
    //            reportLog.setCrAccount(submit.getAccount());
                reportLog.setYmd(smsDetailLog.getYmd());

                //检测reportCache是否已经插入
                Query q = new Query();
                q.addCriteria(Criteria.where("prMsgId").is(reportLog.getPrMsgId()));
                int allTime = 3000;
                long count  = 0;
                while (allTime > 0) {
                    allTime -= 10;
                    count = mongoHelper.count(q, ReportCache.COLLECTION_NAME);
                    if(count>0) break;
                    try { Thread.sleep(10); } catch (InterruptedException ignored) {}
                }
                if(count==0) {
                    producerService.sendMqMessage(RocketMqConstant.SMS_CMPP_REPORT_TOPIC, JSON.toJSONString(reportLog));
                }
            }
        } catch (Exception e) {
            log.error("messageCompleteReject-----submitId="+smsDetailLog.getId()+"----reportCache="+ JSON.toJSONString(reportCache),e);
        }


    }

    @Override
    public void messageExpired(String key, IMessage msgRequest) {
        log.error("messageExpired--------------------SubmitLog.id={}", smsDetailLog.getId());
        CommonThreadPoolFactory.getInstance().getBizPoolExecutor().submit(() -> {
            log.debug("asyncWrite submit expired========> {}", smsDetailLog.toString());
            ProducerService producerService = SpringUtil.getBean(ProducerService.class);

            SubmitRes submitRes = new SubmitRes();
            submitRes.setSmsDetailLogId(smsDetailLog.getId());
            submitRes.setCreateDate(new Date());
            submitRes.setCrMsgId(smsDetailLog.getCrMsgId());
            submitRes.setSubmitStatus(2);
            submitRes.setYmd(smsDetailLog.getYmd());
            log.debug("submitRes,{}", JSON.toJSONString(submitRes));
            producerService.sendDbMessage(submitRes);
//            producerService.sendMessage(RocketMqCommonTopic.PROCESS_LOG_TOPIC + ":" + SubmitRes.COLLECTION_NAME, JSON.toJSONString(submitRes));

			//TODO
            //发送响应更新到统计表中
           /* SendDetailStat sendDetail = new SendDetailStat();
            sendDetail.setId(submitLog.getId());
            sendDetail.setSubmitStatus(2);
            sendDetail.setYmd(submitLog.getYmd());
//            producerService.sendOrderMessage(RocketMqCommonTopic.SEND_DETAIL_TOPIC, JSON.toJSONString(sendDetail),submitLog.getId());
//            producerService.sendMessage(RocketMqCommonTopic.SEND_DETAIL_TOPIC+"_tmp",JSON.toJSONString(sendDetail));
            producerService.sendDbMessage(sendDetail,SendDetailStat.COLLECTION_NAME+"_tmp");*/

            ReportLog reportLog = new ReportLog();
            reportLog.setSmsDetailLogId(smsDetailLog.getId());
            reportLog.setCrMsgId(smsDetailLog.getCrMsgId());
            reportLog.setStat("A:" + _40001.getStatus());
            reportLog.setSubmitTime(new DateTime(smsDetailLog.getSubmitDate()).toString("yyMMddHHmm"));
            reportLog.setDoneTime(new DateTime().toString("yyMMddHHmm"));
            reportLog.setPhoneNo(smsDetailLog.getPhoneNos());
            reportLog.setCreateDate(new Date());
            reportLog.setProviderNo(smsDetailLog.getProvProviderId());
            reportLog.setCstmCustomerCode(smsDetailLog.getCstmCustomerCode());
            reportLog.setYmd(smsDetailLog.getYmd());
            log.debug("messageExpired-----reportLog={}", JSON.toJSONString(reportLog));
            producerService.sendMqMessage(RocketMqConstant.SMS_CMPP_REPORT_TOPIC, JSON.toJSONString(reportLog));
        });
    }

    @Override
    public void sendMessageFailed(IMessage msgRequest) {
        log.error("sendMessageFailed--------------------");
        CommonThreadPoolFactory.getInstance().getBizPoolExecutor().submit(() -> {
            log.debug("asyncWrite submit faild========> {}", smsDetailLog.toString());
            ProducerService producerService = SpringUtil.getBean(ProducerService.class);
            SubmitRes submitRes = new SubmitRes();
            submitRes.setSmsDetailLogId(smsDetailLog.getId());
            submitRes.setCreateDate(new Date());
            submitRes.setCrMsgId(smsDetailLog.getCrMsgId());
            submitRes.setSubmitStatus(1);
            submitRes.setYmd(smsDetailLog.getYmd());
            log.debug("submitRes,{}", JSON.toJSONString(submitRes));
            producerService.sendDbMessage(submitRes);
//            producerService.sendMessage(RocketMqCommonTopic.PROCESS_LOG_TOPIC + ":" + SubmitRes.COLLECTION_NAME, JSON.toJSONString(submitRes));

			//TODO
            //发送响应更新到统计表中
            /*SendDetailStat sendDetail = new SendDetailStat();
            sendDetail.setId(submitLog.getId());
            sendDetail.setSubmitStatus(2);
            sendDetail.setYmd(submitLog.getYmd());
//            producerService.sendOrderMessage(RocketMqCommonTopic.SEND_DETAIL_TOPIC, JSON.toJSONString(sendDetail),submitLog.getId());
//            producerService.sendMessage(RocketMqCommonTopic.SEND_DETAIL_TOPIC+"_tmp",JSON.toJSONString(sendDetail));
            producerService.sendDbMessage(sendDetail,SendDetailStat.COLLECTION_NAME+"_tmp");*/

            ReportLog reportLog = new ReportLog();
            reportLog.setSmsDetailLogId(smsDetailLog.getId());
            reportLog.setCrMsgId(smsDetailLog.getCrMsgId());
            reportLog.setStat("A:" + _40001.getStatus());
            reportLog.setSubmitTime(new DateTime(smsDetailLog.getSubmitDate()).toString("yyMMddHHmm"));
            reportLog.setDoneTime(new DateTime().toString("yyMMddHHmm"));
            reportLog.setPhoneNo(smsDetailLog.getPhoneNos());
            reportLog.setCreateDate(new Date());
            reportLog.setProviderNo(smsDetailLog.getProvProviderId());
            reportLog.setCstmCustomerCode(smsDetailLog.getCstmCustomerCode());
            reportLog.setYmd(smsDetailLog.getYmd());
            log.debug("sendMessageFailed-----reportLog={}", JSON.toJSONString(reportLog));
            producerService.sendMqMessage(RocketMqConstant.SMS_CMPP_REPORT_TOPIC, JSON.toJSONString(reportLog));

        });
    }
}
